package mr

import (
	"io/ioutil"
	"log"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

type Master struct {
	// Your definitions here.
	fileNames []string
	nReduce   int

	// 这里考虑到，map和reduce是要顺序执行的。
	// 而生产出所有的reduce job也是要时间的，我不希望让第一个拿reduce job的worker等。
	// 所以我在master创建的时候就创建好了另一个reduce job line。当map任务全部完成，只需要将jobs指向nextJobs即可。
	jobs     JobLine
	nextJobs JobLine

	isMapDone    bool
	isReduceDone bool
}

// start a thread that listens for RPCs from worker.go
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

func (m *Master) GetJob(request JobRequest, response *JobResponse) error {
	lastFinishJob := request.Job
	if !lastFinishJob.IsEmpty() {
		m.jobs.Ack(lastFinishJob)
	}

	if m.isReduceDone && m.isMapDone { // 程序即将退出
		response.Job = NewExitJob()
		return nil
	}

	isEmpty, isWait := m.jobs.Status()
	if isEmpty {
		if isWait { // 还有worker没有ack
			response.Job = NewWaitJob()
			return nil
		}

		if !m.isMapDone { // map全都完成了
			m.isMapDone = true
			m.jobs, m.nextJobs = m.nextJobs, JobLine{}
			response.Job = m.jobs.PopAndWait()
		} else { // reduce全都完成
			m.isReduceDone = true
			response.Job = NewExitJob()
		}
		return nil
	}

	response.Job = m.jobs.PopAndWait()
	return nil
}

// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
func (m *Master) Done() bool {
	return m.isMapDone && m.isReduceDone
}

// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	log.Println("get files", files)
	jobs := make([]Job, 0, len(files))
	for _, fileName := range files {
		content := getFileContent(fileName)
		if content == "" {
			log.Println(fileName, "content is empty, continue")
			continue
		}
		jobs = append(jobs, NewMapJob(fileName, content))
	}
	jobLine := NewJobLine()
	jobLine.Add(jobs...)
	nextJobs := make([]Job, 0, len(files))
	for i := 0; i < nReduce; i++ {
		nextJobs = append(nextJobs, NewReduceJob(nReduce, i, files))
	}
	nextJobLine := NewJobLine()
	nextJobLine.Add(nextJobs...)
	m := Master{
		fileNames: files,
		nReduce:   nReduce,
		jobs:      *jobLine,
		nextJobs:  *nextJobLine,
	}

	m.server()
	return &m
}

func getFileContent(fileName string) string {
	file, e := os.Open(fileName)
	defer file.Close()
	if e != nil {
		log.Fatalf("get %s file content fail, detail: %s", fileName, e)
		return ""
	}
	//buffer := []byte{}
	content, e := ioutil.ReadAll(file)
	if e != nil {
		log.Fatalf("get %s file content fail, detail: %s", fileName, e)
		return ""
	}
	return string(content)
}
